package org.fanjr.simplify.zk.server;

import java.io.File;
import java.io.IOException;
import java.util.Properties;

import org.apache.zookeeper.server.ServerCnxnFactory;
import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.slf4j.Logger;
import org.springframework.util.StringUtils;

public class ZkServer {

	private static final Logger LOGGER = org.slf4j.LoggerFactory.getLogger(ZkServer.class);

	private ZooKeeperServer zkServer;

	private ServerCnxnFactory cnxnFactory;

	private ZkServer(ZooKeeperServer zkServer, ServerCnxnFactory cnxnFactory) {
		this.zkServer = zkServer;
		this.cnxnFactory = cnxnFactory;
	}

	public static ZkServer runDefault() throws IOException, ConfigException {
		try {
			return runFromProperties(new Properties());
		} catch (IOException | ConfigException e) {
			LOGGER.error("Start zookeeper server error!", e);
			throw e;
		}
	}

	public static ZkServer runFromProperties(Properties properties) throws ConfigException, IOException {
		QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig();
		initDefaultValue("dataDir", "zkData", properties);
		initDefaultValue("clientPort", "2181", properties);
		quorumPeerConfig.parseProperties(properties);
		ServerConfig config = new ServerConfig();
		config.readFrom(quorumPeerConfig);
		return runFromConfig(config);
	}

	public static ZkServer runFromConfig(ServerConfig config) throws IOException {
		LOGGER.info("Starting zookeeper server");
		FileTxnSnapLog txnLog = null;
		ZooKeeperServer zkServer = new ZooKeeperServer();
		txnLog = new FileTxnSnapLog(new File(config.getDataLogDir()), new File(config.getDataDir()));
		txnLog.setServerStats(zkServer.serverStats());
		zkServer.setTxnLogFactory(txnLog);
		zkServer.setTickTime(config.getTickTime());
		zkServer.setMinSessionTimeout(config.getMinSessionTimeout());
		zkServer.setMaxSessionTimeout(config.getMaxSessionTimeout());
		ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
		cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());
		try {
			cnxnFactory.startup(zkServer);
		} catch (InterruptedException e) {
			// skip
		}
		return new ZkServer(zkServer, cnxnFactory);
	}

	public void destroy() {
		if (cnxnFactory != null) {
			cnxnFactory.shutdown();
		}
	}

	public ZooKeeperServer getZkServer() {
		return zkServer;
	}

	private static void initDefaultValue(String key, String defaultValue, Properties properties) {
		if (!StringUtils.hasText((String) properties.get(key))) {
			properties.put(key, defaultValue);
		}
	}

}
